package org.ms.mconf.sdk.service.redis;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.micro.URL;
import org.micro.extension.Extension;
import org.ms.mconf.sdk.NotifyListener;
import org.ms.mconf.sdk.Constants;
import org.ms.mconf.sdk.exception.MicroConfigException;
import org.ms.mconf.sdk.model.*;
import org.ms.mconf.sdk.service.ops.IAvcService;
import org.ms.mconf.sdk.service.ops.INodeService;
import org.ms.mconf.sdk.service.support.FailBackService;
import org.ms.mconf.sdk.service.support.ChangeLog;
import org.ms.mconf.sdk.service.support.StoreEntry;
import org.ms.mconf.sdk.service.type.ActionType;
import org.ms.mconf.sdk.service.type.DataType;
import redis.clients.jedis.*;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * The Redis Config Service
 *
 * @author lry
 */
@Slf4j
@Extension("redis")
public class RedisService extends FailBackService {

    private JedisPool jedisPool;
    private final Map<String, JedisPubSub> jedisPubSubMap = new ConcurrentHashMap<>();

    // ========= Node Service

    @Override
    public void initialize(URL url) throws MicroConfigException {
        super.initialize(url);

        // Redis Pool Config
        JedisPoolConfig jedisPoolConfig = JSON.parseObject(JSON.toJSONString(url.getParameters()), JedisPoolConfig.class);
        log.info("The Redis Pool Config is: {}", jedisPoolConfig.toString());

        int timeout = url.getParameter(Constants.TIMEOUT_KEY, 0);
        String password = url.getParameter(Constants.PASSWORD_KEY);
        int database = url.getParameter(Constants.DATABASE_KEY, -1);
        if (timeout <= 0 || StringUtils.isBlank(password) || database < 0) {
            jedisPool = new JedisPool(jedisPoolConfig, url.getHost(), url.getPort());
        } else {
            jedisPool = new JedisPool(jedisPoolConfig, url.getHost(), url.getPort(), timeout, password, database);
        }
    }

    @Override
    public void destroy() throws MicroConfigException {
        super.destroy();

        if (jedisPool != null && !jedisPool.isClosed()) {
            jedisPool.close();
        }
    }

    // ========= Config Service

    @Override
    public void subscribe(NotifyListener<MetaMultiDataConfig> listener, String application, String... configs) throws MicroConfigException {
        if (configs == null || configs.length == 0) {
            // build match key
            MetaConfig metaConfig = new MetaConfig(this.node, this.env, application, Constants.MATCH_ALL, this.version);
            String matchKey = MetaConfig.buildChannel(this.group, metaConfig);
            if (jedisPubSubMap.containsKey(matchKey)) {
                return;
            }

            // calculation patterns channel
            JedisPubSub jedisPubSub;
            jedisPubSubMap.put(matchKey, jedisPubSub = new JedisPubSub() {
                @Override
                public void onPMessage(String pattern, String channel, String message) {
                    log.debug("The subscribe all notify, channel={}, message={}", pattern, channel, message);
                    listener.notify(parseAndBuild(channel, message));
                }
            });

            // async subscribe
            subscribeThreadPoolExecutor.execute(() -> {
                try (Jedis jedis = jedisPool.getResource()) {
                    log.info("The match subscribe channel: {}", matchKey);
                    jedis.psubscribe(jedisPubSub, matchKey);
                }
            });
        } else {
            // subscribe multiple channel
            JedisPubSub jedisPubSub = new JedisPubSub() {
                @Override
                public void onMessage(String channel, String message) {
                    log.debug("The subscribe[{}] notify, channel={}, message={}", Arrays.toString(configs), channel, message);
                    listener.notify(parseAndBuild(channel, message));
                }
            };

            // calculation channel keys
            List<String> channelKeys = new ArrayList<>();
            for (String config : configs) {
                MetaConfig metaConfig = new MetaConfig(this.node, this.env, application, config, this.version);

                String channel = MetaConfig.buildChannel(this.group, metaConfig);
                if (jedisPubSubMap.containsKey(channel)) {
                    return;
                }
                channelKeys.add(channel);
                jedisPubSubMap.put(channel, jedisPubSub);
            }
            if (channelKeys.isEmpty()) {
                return;
            }

            // async subscribe
            subscribeThreadPoolExecutor.execute(() -> {
                try (Jedis jedis = jedisPool.getResource()) {
                    log.info("The subscribe channels: {}", channelKeys);
                    jedis.subscribe(jedisPubSub, channelKeys.toArray(new String[channelKeys.size()]));
                }
            });
        }

        // the pull config and notify listener when first subscribe
        List<MetaMultiDataConfig> metaMultiDataConfigs = this.pull(application, configs);
        if (metaMultiDataConfigs == null || metaMultiDataConfigs.isEmpty()) {
            return;
        }
        for (MetaMultiDataConfig metaMultiDataConfig : metaMultiDataConfigs) {
            if (metaMultiDataConfig.getData().isEmpty()) {
                continue;
            }

            String channelKey = MetaConfig.buildChannel(this.group, metaMultiDataConfig.getMeta());
            subscribeMetaMultiDataConfigs.computeIfAbsent(channelKey, k -> new HashSet<>()).add(metaMultiDataConfig);
        }
        listener.notify(metaMultiDataConfigs);
    }

    @Override
    public void unsubscribe(String application, String... configs) throws MicroConfigException {
        if (configs == null || configs.length == 0) {
            MetaConfig metaConfig = new MetaConfig(this.node, this.env, application, Constants.MATCH_ALL, this.version);

            String matchKey = MetaConfig.buildChannel(this.group, metaConfig);
            JedisPubSub jedisPubSub = jedisPubSubMap.get(matchKey);
            if (jedisPubSub == null) {
                return;
            }
            jedisPubSub.punsubscribe(matchKey);
            jedisPubSubMap.remove(matchKey);
            log.info("The unsubscribe match channel: {}", matchKey);
        } else {
            for (String config : configs) {
                MetaConfig metaConfig = new MetaConfig(this.node, this.env, application, config, this.version);

                String channelKey = MetaConfig.buildChannel(this.group, metaConfig);
                JedisPubSub jedisPubSub = jedisPubSubMap.get(channelKey);
                if (jedisPubSub == null) {
                    return;
                }
                jedisPubSub.unsubscribe(channelKey);
                jedisPubSubMap.remove(channelKey);
                log.info("The unsubscribe channels: {}", channelKey);
            }
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public List<MetaMultiDataConfig> pullCall(String application, String... configs) throws MicroConfigException {
        // the build key-config
        List<String> keys = new ArrayList<>();
        if (configs == null || configs.length == 0) {
            Set<String> keySet;
            try (Jedis jedis = jedisPool.getResource()) {
                MetaConfig metaConfig = new MetaConfig(this.node, this.env, application, Constants.MATCH_ALL, this.version);
                String matchKey = MetaConfig.buildMatchKey(this.group, metaConfig);
                keySet = jedis.keys(matchKey);
            }

            if (keySet == null || keySet.isEmpty()) {
                return Collections.emptyList();
            }
            keys.addAll(keySet);
        } else {
            for (String config : configs) {
                MetaConfig metaConfig = new MetaConfig(this.node, this.env, application, config, this.version);
                keys.add(MetaConfig.buildData(this.group, metaConfig));
            }
        }

        // get all data
        Map<String, Set<String>> queryMap = this.batchGetAll(keys);
        List<MetaMultiDataConfig> pullResultList = new ArrayList<>();
        for (Map.Entry<String, Set<String>> entry : queryMap.entrySet()) {
            MetaConfig metaConfig = MetaConfig.parse(entry.getKey());
            List<DataConfig> dataConfigs = entry.getValue().stream().map(DataConfig::parseDataValue).collect(Collectors.toList());

            pullResultList.add(new MetaMultiDataConfig(metaConfig, dataConfigs));
        }

        return pullResultList;
    }


    // =========== Module 1 Ops: Config Data Resource


    @Override
    public boolean publish(MetaMultiDataConfig metaMultiDataConfig) throws MicroConfigException {
        super.publish(metaMultiDataConfig);

        // build key
        String key = MetaConfig.buildData(this.group, metaMultiDataConfig.getMeta());

        // query config list
        Map<String, String> queryHash;
        try (Jedis jedis = jedisPool.getResource()) {
            queryHash = jedis.hgetAll(key);
        }
        Map<String, DataConfig> oldHash = new HashMap<>();
        if (queryHash != null && !queryHash.isEmpty()) {
            for (Map.Entry<String, String> entry : queryHash.entrySet()) {
                oldHash.put(entry.getKey(), DataConfig.parseDataValue(entry.getValue()));
            }
        }

        // grouping statistics
        List<DataConfig> addDataConfig = new ArrayList<>();
        List<DataConfig> updateDataConfig = new ArrayList<>();
        List<DataConfig> oldDataConfig = new ArrayList<>();

        // build hash key-value list
        Map<String, String> hash = new HashMap<>(metaMultiDataConfig.getData().size());
        for (DataConfig dataConfig : metaMultiDataConfig.getData()) {
            hash.put(dataConfig.getKey(), DataConfig.buildDataValue(dataConfig));
            // grouping statistics
            if (oldHash.containsKey(dataConfig.getKey())) {
                updateDataConfig.add(dataConfig);
                oldDataConfig.add(oldHash.get(dataConfig.getKey()));
            } else {
                addDataConfig.add(dataConfig);
            }
        }

        // publish
        try (Jedis jedis = jedisPool.getResource()) {
            log.info("The publish config, key={}, value={}", key, hash);
            jedis.hmset(key, hash);
        }

        // change log
        if (!addDataConfig.isEmpty()) {
            this.doChangeLog(ActionType.ADD, ChangeLog.buildAdds(metaMultiDataConfig.getMeta(), addDataConfig));
        }
        if (!updateDataConfig.isEmpty()) {
            this.doChangeLog(ActionType.UPDATE, ChangeLog.buildUpdates(metaMultiDataConfig.getMeta(), updateDataConfig, oldDataConfig));
        }

        return true;
    }

    @Override
    public boolean delete(MetaConfig metaConfig, String... keys) throws MicroConfigException {
        boolean flag;
        List<ChangeLog> changeLogs = new ArrayList<>();
        if (keys != null && keys.length > 0) {
            if (StringUtils.isBlank(metaConfig.getNode()) ||
                    StringUtils.isBlank(metaConfig.getEnvironment()) ||
                    StringUtils.isBlank(metaConfig.getApplication()) ||
                    StringUtils.isBlank(metaConfig.getConfig()) ||
                    Constants.MATCH_ALL.equals(metaConfig.getNode()) ||
                    Constants.MATCH_ALL.equals(metaConfig.getEnvironment()) ||
                    Constants.MATCH_ALL.equals(metaConfig.getApplication()) ||
                    Constants.MATCH_ALL.equals(metaConfig.getConfig()) ||
                    Constants.MATCH_ALL.equals(metaConfig.getVersion())) {
                throw new IllegalArgumentException(metaConfig.toString());
            }

            // build key
            String key = MetaConfig.buildData(this.group, metaConfig);

            // query config
            List<String> values;
            try (Jedis jedis = jedisPool.getResource()) {
                values = jedis.hmget(key, keys);
            }
            if (values == null || values.isEmpty()) {
                throw new IllegalArgumentException("The not found values by: " + Arrays.toString(keys));
            }
            List<DataConfig> dataConfigs = values.stream().map(DataConfig::parseDataValue).collect(Collectors.toList());
            changeLogs.add(ChangeLog.buildAddOrDelete(metaConfig, dataConfigs));

            // delete config
            try (Jedis jedis = jedisPool.getResource()) {
                flag = jedis.hdel(key, keys) > 0L;
            }
        } else {
            // build key
            Set<String> deleteKeys = new HashSet<>();
            if (StringUtils.isBlank(metaConfig.getNode()) ||
                    StringUtils.isBlank(metaConfig.getEnvironment()) ||
                    StringUtils.isBlank(metaConfig.getApplication()) ||
                    StringUtils.isBlank(metaConfig.getConfig()) ||
                    StringUtils.isBlank(metaConfig.getVersion())) {
                String matchKey = MetaConfig.buildMatchKey(this.group, metaConfig);

                try (Jedis jedis = jedisPool.getResource()) {
                    deleteKeys = jedis.keys(matchKey);
                }
            } else {
                deleteKeys.add(MetaConfig.buildData(this.group, metaConfig));
            }
            if (deleteKeys == null || deleteKeys.isEmpty()) {
                throw new MicroConfigException("The NotFound delete keys");
            }

            // batch query
            Map<String, Set<String>> queryMap = this.batchGetAll(deleteKeys);
            for (Map.Entry<String, Set<String>> entry : queryMap.entrySet()) {
                List<DataConfig> dataConfigList = new ArrayList<>();
                for (String value : entry.getValue()) {
                    dataConfigList.add(DataConfig.parseDataValue(value));
                }
                changeLogs.add(ChangeLog.buildAddOrDelete(MetaConfig.parse(entry.getKey()), dataConfigList));
            }

            // delete config
            try (Jedis jedis = jedisPool.getResource()) {
                flag = jedis.del(deleteKeys.toArray(new String[deleteKeys.size()])) > 0L;
            }
        }

        if (flag) {
            // change log
            this.doChangeLog(ActionType.DELETE, changeLogs);
        }

        return flag;
    }

    @SuppressWarnings("unchecked")
    @Override
    public List<MetaDataConfig> search(MetaConfig metaConfig, String keyword) throws MicroConfigException {
        // build match key
        String matchKey = MetaConfig.buildMatchKey(this.group, metaConfig);

        // query config data keys
        Set<String> keys;
        try (Jedis jedis = jedisPool.getResource()) {
            keys = jedis.keys(matchKey);
        }
        if (keys == null || keys.isEmpty()) {
            return Collections.emptyList();
        }

        // query config data values
        Map<String, Set<String>> queryMap = this.batchGetAll(keys);
        // parse config data value
        List<MetaDataConfig> metaDataConfigs = new ArrayList<>();
        for (Map.Entry<String, Set<String>> entry : queryMap.entrySet()) {
            MetaConfig tempMetaConfig = MetaConfig.parse(entry.getKey());

            for (String value : entry.getValue()) {
                MetaDataConfig metaDataConfig = new MetaDataConfig(tempMetaConfig, DataConfig.parseDataValue(value));
                if (this.matchKeyword(metaDataConfig, keyword)) {
                    metaDataConfigs.add(metaDataConfig);
                }
            }
        }

        return metaDataConfigs;
    }

    @Override
    public MetaDataConfig query(MetaConfig metaConfig, String key) throws MicroConfigException {
        // build match key
        String queryKey = MetaConfig.buildKey(this.group, metaConfig);

        // query config data keys
        String value;
        try (Jedis jedis = jedisPool.getResource()) {
            value = jedis.hget(queryKey, key);
        }

        if (value == null || value.length() == 0) {
            return null;
        }

        return new MetaDataConfig(metaConfig, DataConfig.parseDataValue(value));
    }

    @Override
    public void subscribe(NotifyListener<MetaDataConfig> listener) throws MicroConfigException {
        String matchKey = MetaConfig.buildChannel(this.group, new MetaConfig(Constants.MATCH_ALL,
                Constants.MATCH_ALL, Constants.MATCH_ALL, Constants.MATCH_ALL, Constants.MATCH_ALL));

        // patterns subscribe
        JedisPubSub jedisPubSub;
        jedisPubSubMap.put(matchKey, jedisPubSub = new JedisPubSub() {
            @Override
            public void onPMessage(String pattern, String channel, String message) {
                // the parse message
                MetaMultiDataConfig metaMultiDataConfig = MetaMultiDataConfig.parseMetaMultiDataConfig(channel, message);
                Set<MetaDataConfig> metaDataConfigs = new HashSet<>(metaMultiDataConfig.getData().size());
                for (DataConfig dataConfig : metaMultiDataConfig.getData()) {
                    metaDataConfigs.add(new MetaDataConfig(metaMultiDataConfig.getMeta(), dataConfig));
                }

                // the calculation ActionType
                ActionType actionType = DataConfig.parseActionType(message);
                if (ActionType.ADD == actionType || ActionType.UPDATE == actionType) {
                    subscribeMetaDataConfigs.put(channel, metaDataConfigs);
                } else if (ActionType.DELETE == actionType) {
                    subscribeMetaDataConfigs.remove(channel);
                } else {
                    return;
                }
                if (subscribeMetaDataConfigs.isEmpty()) {
                    return;
                }

                // the notify config
                List<MetaDataConfig> results = new ArrayList<>();
                for (Set<MetaDataConfig> subscribeMetaDataConfig : subscribeMetaDataConfigs.values()) {
                    results.addAll(subscribeMetaDataConfig);
                }
                listener.notify(results);
            }
        });

        // submit executor
        subscribeThreadPoolExecutor.execute(() -> {
            try (Jedis jedis = jedisPool.getResource()) {
                log.info("The match subscribe all channel: {}", matchKey);
                jedis.psubscribe(jedisPubSub, matchKey);
            }
        });

        // the search config and notify listener when first subscribe
        List<MetaDataConfig> metaDataConfigs = this.search(new MetaConfig(), null);
        if (metaDataConfigs != null && !metaDataConfigs.isEmpty()) {
            for (MetaDataConfig metaConfig : metaDataConfigs) {
                String channelKey = MetaConfig.buildChannel(this.group, metaConfig.getMetaConfig());
                subscribeMetaDataConfigs.computeIfAbsent(channelKey, k -> new HashSet<>()).add(metaConfig);
            }
            listener.notify(metaDataConfigs);
        }
    }

    @Override
    public void unsubscribe() throws MicroConfigException {
        String matchKey = MetaConfig.buildChannel(this.group, new MetaConfig(Constants.MATCH_ALL,
                Constants.MATCH_ALL, Constants.MATCH_ALL, Constants.MATCH_ALL, Constants.MATCH_ALL));

        JedisPubSub jedisPubSub = jedisPubSubMap.get(matchKey);
        if (jedisPubSub == null) {
            return;
        }

        jedisPubSub.punsubscribe(matchKey);
        jedisPubSubMap.remove(matchKey);
        log.info("The unsubscribe match all channel: {}", matchKey);
    }


    // =========== Module 2 Ops: Node Resource


    @Override
    public INodeService nodeService() {
        return nodeService;
    }

    private INodeService nodeService = new INodeService() {
        @Override
        public boolean publish(NodeConfig nodeConfig) throws MicroConfigException {
            if (nodeConfig.getId() == null || nodeConfig.getId().length() == 0) {
                throw new IllegalArgumentException("The parameter id can not be empty or length 0");
            }

            String resourceKey = NodeConfig.buildNodeKey(group);
            try (Jedis jedis = jedisPool.getResource()) {
                jedis.hset(resourceKey, nodeConfig.getId(), NodeConfig.buildValue(nodeConfig));
                return true;
            }
        }

        @Override
        public boolean delete(String node) throws MicroConfigException {
            String resourceKey = NodeConfig.buildNodeKey(group);
            try (Jedis jedis = jedisPool.getResource()) {
                jedis.hdel(resourceKey, node);
                return true;
            }
        }

        @Override
        public Set<NodeConfig> query() throws MicroConfigException {
            String resourceKey = NodeConfig.buildNodeKey(group);

            // get config
            Map<String, String> hash;
            try (Jedis jedis = jedisPool.getResource()) {
                hash = jedis.hgetAll(resourceKey);
            }
            if (hash == null || hash.isEmpty()) {
                return Collections.emptySet();
            }

            // parse config
            Set<NodeConfig> nodeConfigs = new HashSet<>();
            for (Map.Entry<String, String> entry : hash.entrySet()) {
                nodeConfigs.add(NodeConfig.parseValue(entry.getValue()));
            }

            return nodeConfigs;
        }
    };


    // =========== Module 3 Ops: Environment Resource


    @Override
    public INodeService envService() {
        return envService;
    }

    private INodeService envService = new INodeService() {
        @Override
        public boolean publish(NodeConfig nodeConfig) throws MicroConfigException {
            if (nodeConfig.getId() == null || nodeConfig.getId().length() == 0) {
                throw new IllegalArgumentException("The parameter id can not be empty or length 0");
            }

            String resourceKey = NodeConfig.buildEnvKey(group);
            try (Jedis jedis = jedisPool.getResource()) {
                jedis.hset(resourceKey, nodeConfig.getId(), NodeConfig.buildValue(nodeConfig));
                return true;
            }
        }

        @Override
        public boolean delete(String node) throws MicroConfigException {
            String resourceKey = NodeConfig.buildEnvKey(group);

            try (Jedis jedis = jedisPool.getResource()) {
                jedis.hdel(resourceKey, node);
                return true;
            }
        }

        @Override
        public Set<NodeConfig> query() throws MicroConfigException {
            String resourceKey = NodeConfig.buildEnvKey(group);

            // get config
            Map<String, String> hash;
            try (Jedis jedis = jedisPool.getResource()) {
                hash = jedis.hgetAll(resourceKey);
            }
            if (hash == null || hash.isEmpty()) {
                return Collections.emptySet();
            }

            // parse config
            Set<NodeConfig> nodeConfigs = new HashSet<>();
            for (Map.Entry<String, String> entry : hash.entrySet()) {
                nodeConfigs.add(NodeConfig.parseValue(entry.getValue()));
            }

            return nodeConfigs;
        }
    };


    // =========== Module 4 Ops: Application-Config-Version Resource


    @Override
    public IAvcService acvService() {
        return avcService;
    }

    private IAvcService avcService = new IAvcService() {
        @Override
        public boolean publish(NodeMultiConfig nodeMultiConfig) throws MicroConfigException {
            NodeConfig applicationNodeConfig = nodeMultiConfig.getApplication();
            if (applicationNodeConfig.getId() == null || applicationNodeConfig.getId().length() == 0) {
                throw new IllegalArgumentException("The application id can not be empty or length 0");
            }
            NodeConfig configNodeConfig = nodeMultiConfig.getConfig();
            if (configNodeConfig.getId() == null || configNodeConfig.getId().length() == 0) {
                throw new IllegalArgumentException("The config id can not be empty or length 0");
            }
            NodeConfig versionNodeConfig = nodeMultiConfig.getVersion();
            if (versionNodeConfig.getId() == null || versionNodeConfig.getId().length() == 0) {
                throw new IllegalArgumentException("The version id can not be empty or length 0");
            }

            String resourceKey = NodeMultiConfig.buildKey(group);
            String resourceField = NodeMultiConfig.buildField(nodeMultiConfig);
            String resourceValue = NodeMultiConfig.buildValue(nodeMultiConfig);

            try (Jedis jedis = jedisPool.getResource()) {
                jedis.hset(resourceKey, resourceField, resourceValue);
                return true;
            }
        }

        @Override
        public boolean delete(NodeMultiConfig nodeMultiConfig) throws MicroConfigException {
            String resourceKey = NodeMultiConfig.buildKey(group);
            String resourceField = NodeMultiConfig.buildField(nodeMultiConfig);

            try (Jedis jedis = jedisPool.getResource()) {
                jedis.hdel(resourceKey, resourceField);
                return true;
            }
        }

        @Override
        public Set<NodeMultiConfig> query() throws MicroConfigException {
            String resourceKey = NodeMultiConfig.buildKey(group);

            // get config
            Map<String, String> hash;
            try (Jedis jedis = jedisPool.getResource()) {
                hash = jedis.hgetAll(resourceKey);
            }
            if (hash == null || hash.isEmpty()) {
                return Collections.emptySet();
            }

            // parse config
            Set<NodeMultiConfig> nodeMultiConfigs = new HashSet<>();
            for (Map.Entry<String, String> entry : hash.entrySet()) {
                nodeMultiConfigs.add(NodeMultiConfig.parseValue(entry.getValue()));
            }

            return nodeMultiConfigs;
        }

        @Override
        public Set<NodeConfig> select() throws MicroConfigException {
            return null;
        }
    };


    // =========== Module 5 Ops: Config Data Transfer


    @Override
    public void clear() throws MicroConfigException {
        try (Jedis jedis = jedisPool.getResource()) {
            String matchKey = this.group + Constants.DELIMITER + Constants.MATCH_ALL;
            Set<String> keys = jedis.keys(matchKey);
            if (keys == null || keys.isEmpty()) {
                return;
            }

            jedis.del(keys.toArray(new String[keys.size()]));
            log.info("The clear all config data by keys: {}", keys);
        }
    }

    @Override
    public List<String> backup() throws MicroConfigException {
        List<String> storeEntries = new ArrayList<>();

        // calculation node config
        Set<NodeConfig> nodeConfigs = this.nodeService().query();
        if (nodeConfigs != null && !nodeConfigs.isEmpty()) {
            for (NodeConfig nodeConfig : nodeConfigs) {
                storeEntries.add(JSON.toJSONString(new StoreEntry<>(DataType.NODE_CONFIG, nodeConfig)));
            }
        }

        // calculation env config
        Set<NodeConfig> envConfigs = this.envService().query();
        if (envConfigs != null && !envConfigs.isEmpty()) {
            for (NodeConfig envConfig : envConfigs) {
                storeEntries.add(JSON.toJSONString(new StoreEntry<>(DataType.ENV_CONFIG, envConfig)));
            }
        }

        // calculation acv config
        Set<NodeMultiConfig> acvConfigs = this.acvService().query();
        if (acvConfigs != null && !acvConfigs.isEmpty()) {
            for (NodeMultiConfig nodeMultiConfig : acvConfigs) {
                storeEntries.add(JSON.toJSONString(new StoreEntry<>(DataType.ACV_CONFIG, nodeMultiConfig)));
            }
        }

        // calculation config data
        List<MetaDataConfig> metaDataConfigs = this.search(new MetaConfig(), null);
        if (metaDataConfigs != null && !metaDataConfigs.isEmpty()) {
            for (MetaDataConfig metaDataConfig : metaDataConfigs) {
                storeEntries.add(JSON.toJSONString(new StoreEntry<>(DataType.DATA_CONFIG, metaDataConfig)));
            }
        }

        return storeEntries;
    }

    @Override
    public void recovery(List<String> data) throws MicroConfigException {
        for (String datum : data) {
            StoreEntry storeEntry = JSON.parseObject(datum, StoreEntry.class);
            if (DataType.NODE_CONFIG == storeEntry.getDataType()) {

            } else if (DataType.ENV_CONFIG == storeEntry.getDataType()) {

            } else if (DataType.ACV_CONFIG == storeEntry.getDataType()) {

            } else if (DataType.DATA_CONFIG == storeEntry.getDataType()) {

            } else {
                log.warn("The Illegal DataType: " + storeEntry.getDataType());
            }
        }
    }


    // ======== Internal use


    private Map<String, Set<String>> batchGetAll(Collection<String> collection) {
        Map<String, Set<String>> result = new HashMap<>();
        Map<String, Response<Map<String, String>>> map = new HashMap<>();
        try (Jedis jedis = jedisPool.getResource()) {
            Pipeline pipeline = jedis.pipelined();
            for (String key : collection) {
                map.put(key, pipeline.hgetAll(key));
            }
            pipeline.sync();
            try {
                pipeline.close();
            } catch (IOException e) {
                log.error("The pipeline close is Exception", e);
            }
        }

        for (Map.Entry<String, Response<Map<String, String>>> entry : map.entrySet()) {
            Collection<String> values = entry.getValue().get().values();
            if (values.isEmpty()) {
                continue;
            }

            result.put(entry.getKey(), new HashSet<>(values));
        }

        return result;
    }

    private void doChangeLog(ActionType actionType, List<ChangeLog> changeLogs) {
        for (ChangeLog changeLog : changeLogs) {
            // build key
            String channelKey = MetaConfig.buildChannel(this.group, changeLog.getMetaConfig());
            String message = DataConfig.buildMessage(actionType, changeLog.getDataConfigs());

            try (Jedis jedis = jedisPool.getResource()) {
                jedis.publish(channelKey, message);
                log.info("The publish notify channel, key={}, message={}", channelKey, message);
            }
        }
    }

    private boolean matchKeyword(MetaDataConfig metaDataConfig, String keyword) {
        if (StringUtils.isBlank(keyword)) {
            return true;
        }

        MetaConfig metaConfig = metaDataConfig.getMetaConfig();
        if (metaConfig.getEnvironment().contains(keyword) ||
                metaConfig.getApplication().contains(keyword) ||
                metaConfig.getConfig().contains(keyword) ||
                metaConfig.getVersion().contains(keyword)) {
            return true;
        }

        DataConfig dataConfig = metaDataConfig.getDataConfig();
        return dataConfig.getKey().contains(keyword) ||
                dataConfig.getValue().toString().contains(keyword) ||
                dataConfig.getIntroduce().contains(keyword);
    }

    private List<MetaMultiDataConfig> parseAndBuild(String channel, String message) {
        ActionType actionType = DataConfig.parseActionType(message);
        MetaMultiDataConfig metaMultiDataConfig = MetaMultiDataConfig.parseMetaMultiDataConfig(channel, message);
        if (ActionType.ADD == actionType || ActionType.UPDATE == actionType) {
            Set<MetaMultiDataConfig> metaMultiDataConfigs = subscribeMetaMultiDataConfigs.computeIfAbsent(channel, k -> new HashSet<>());
            metaMultiDataConfigs.add(metaMultiDataConfig);
        } else if (ActionType.DELETE == actionType) {
            subscribeMetaMultiDataConfigs.remove(channel);
        } else {
            throw new IllegalArgumentException(actionType.name());
        }

        List<MetaMultiDataConfig> results = new ArrayList<>();
        for (Set<MetaMultiDataConfig> subscribeMetaDataConfig : subscribeMetaMultiDataConfigs.values()) {
            results.addAll(subscribeMetaDataConfig);
        }

        return results;
    }

}
